package com.young.async;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.io.UnsupportedEncodingException;

/**
 * 异步消息通常用在对响应时间铭感的业务场景，即发送端不能容忍长时间地等待Broker的响应
 */
public class AsyncProducer {
    public static void main(String[] args) throws MQClientException, UnsupportedEncodingException, RemotingException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer("async_producer_group");
        producer.setNamesrvAddr("192.168.188.101:9876;192.168.188.102:9876;");
        producer.start();
        for (int i = 0; i < 10; i++) {
            Message message = new Message("async_topic", "TagA", ("this is async message" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            producer.send(message, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.println("消息发送成功 MsgId：" + sendResult.getMsgId());
                }

                @Override
                public void onException(Throwable throwable) {
                    System.out.println("发送失败.....");
                    throwable.printStackTrace();
                }
            });
        }
        // 让线程睡一会，等待消息发送完成，关闭消费者
        Thread.sleep(1000);
        producer.shutdown();

    }
}
